package web.service.rest;

import cn.edu.hit.OrbitTrack;
import cn.edu.hit.tendency.DataPoint;
import cn.edu.hit.tendency.RegressionLine;
import cn.edu.hit.tendency.TendencySourceQueue;
import cn.edu.hit.core.Pair;
import cn.edu.hit.core.combine.CombineLoader;
import cn.edu.hit.core.message.MessageLoader;
import cn.edu.hit.kg.Neo4jConfigurationUtil;
//import cn.edu.hit.kg.NeoDriver;
import cn.edu.hit.kg.SignalUtil;
import cn.edu.hit.kg.data.GraphData;
import cn.edu.hit.kg.entity.Fault;
import cn.edu.hit.kg.entity.Signal;
import cn.edu.hit.mongodb.DBHelper;
import cn.edu.hit.mongodb.Utils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Sorts;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import entity.SignalGroup;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.commons.collections.ListUtils;
import org.apache.commons.lang3.StringUtils;
import cn.edu.hit.core.Document;
import org.bson.types.ObjectId;

import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.*;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import static cn.edu.hit.core.combine.CombineAdapter.*;
import static cn.edu.hit.core.message.MessageAdapter.*;

import static web.service.rest.MqConfigure.*;


@Path("/work")
public class WorkService {

    private MessageLoader messageLoader = MessageLoader.getInstance();
    private CombineLoader combineLoader = CombineLoader.getInstance();
    private DBHelper dbHelper = new DBHelper();

    final static Map<String, Boolean> isRunMap = Collections.singletonMap("isRun", true);

    public WorkService() {
//        GraphData.setProvider();
    }

    @Context
    HttpServletRequest request;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Boolean> isRun() {
        return isRunMap;
    }

    /**
     * 根据配置生成信号组并存入数据库
     *
     * @param unit      机组id
     * @param timestamp 时间戳
     * @return 生成的信号组
     */
    @GET
    @Path("/signal/{unit}/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map> generateSignals(@PathParam("unit") String unit, @PathParam("timestamp") long timestamp) {
        List<Document> documentStream = dbHelper.loadConfigure(unit, "signal");

        List<Map> results = messageLoader.transfer(documentStream.stream().map(d -> (Map) d),
                d -> d.get(ID).toString(), d -> d.get(DATA_SOURCE).toString(), timestamp)
                .collect(Collectors.toList());

        dbHelper.persistMessage(results);

        return results;
    }

    @GET
    @Path("/mapping/{chainId}/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map> doMapping(@PathParam("chainId") String chainId, @PathParam("timestamp") long timestamp) {
        List<Document> documentStream = dbHelper.loadConfigure(chainId, "feature");

        MongoDatabase database = Utils.getDatabase("fault");
        MongoCollection<Document> features = database.getCollection("features", Document.class);
        Map<String, Object> rawFeatures = StreamSupport.stream(
                features.find(Filters.eq("timestamp", timestamp))
                        .spliterator(), true).collect(Collectors.groupingBy(
                document -> ((Document) document).getString(PORT_ID),
                Collectors.collectingAndThen(Collectors.toSet(), set -> set.stream()
//                        !!!---2018-08-24----!!!
//                        .map(doc->doc.get(VALUE))
                        .findFirst().get())
        ));
//        System.out.println(rawFeatures);

        List<Map> results = combineLoader.transfer(documentStream.stream().map(d -> (Map) d),
                d -> d.get(ID).toString(), d -> d.get(COMBINE_STRATEGY).toString(), Pair.of(timestamp, rawFeatures)).collect(Collectors.toList());
        return results;
    }

    @GET
    @Path("/chain/{chainId}/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map> doChain(@PathParam("chainId") String chainId, @PathParam("timestamp") long timestamp) {
        List<Map> docments = this.doMapping(chainId, timestamp);
        dbHelper.persistMessage(docments);
        return docments;
    }

    @GET
    @Path("/kg/signal/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Signal> getKnowledgeSignals(@PathParam("timestamp") long timestamp) {
        MongoDatabase database = Utils.getDatabase("fault");
        MongoCollection<Document> features = database.getCollection("features", Document.class);

        List<Signal> rawFeatures = StreamSupport.stream(
                features.find(Filters.and(Filters.eq("timestamp", timestamp), Filters.exists("signalType")))
                        .spliterator(), true)
                .map(SignalUtil::transferFeatureToKG).collect(Collectors.toList());

        List<Signal> rawSignals = StreamSupport.stream(
                features.find(Filters.and(Filters.eq("timestamp", timestamp), Filters.not(Filters.exists("signalType"))))
                        .spliterator(), true)
                .map(SignalUtil::transferSignalToKG).collect(Collectors.toList());

        return ListUtils.union(rawFeatures, rawSignals);
    }


    @GET
    @Path("/kg/match/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map> matchKnowledge(@PathParam("timestamp") long timestamp) {
        List<Signal> rawFeatures = this.getKnowledgeSignals(timestamp);

        GraphData graphData = GraphData.getInstance();

//        try (NeoDriver driver = new NeoDriver()){
        return
//                    driver.query("match (n:Fault) where n.outId is not null return n ")
//                    .list(record -> new Fault(record.get("n").asNode()))
//                    .stream()
                graphData.getIdSetByLabel("Fault").stream().map(graphData::getById)
                        .filter(node -> node.get("outId") != null)
                        .map(node1 -> Fault.loadById(node1.id()))
                        .map(f -> {
                            f.matchSignal(rawFeatures, timestamp);
                            return f.transferFaultToMap();
                        })
                        .filter(map -> (boolean) map.get("isMatched") || (boolean) map.get("isUnknown"))
                        .collect(Collectors.toList());
//        }
    }

    @POST
    @Path("/kg/match/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    @Consumes({MediaType.APPLICATION_FORM_URLENCODED})
    public List<Map> matchKnowledgeWithConditions(@PathParam("timestamp") long timestamp, @FormParam("content") String argsJson) {
        JSONArray otherFeatures = JSONArray.fromObject(argsJson);

        List<Signal> rawFeatures = this.lastKnowledgeSignals(timestamp);
        otherFeatures.forEach(otherFeature -> {
            JSONObject feature = (JSONObject) otherFeature;
            Signal signal = new Signal();
            signal.setValue(feature.opt("value"));
            signal.setComponent(feature.optJSONObject("component"));
            signal.setFeatureType(feature.optString("featureType"));
            rawFeatures.add(signal);
        });

        System.out.println("In matchKnowledgeWithConditions()," + rawFeatures);
        GraphData graphData = GraphData.getInstance();

//        try (NeoDriver driver = new NeoDriver()){
        return
                graphData.getIdSetByLabel("Fault").stream().map(graphData::getById)
                        .filter(node -> node.get("outId") != null)
                        .map(node1 -> Fault.loadById(node1.id()))
//                    driver.query("match (n:Fault) where n.outId is not null return n ")
//                    .list(record -> new Fault(record.get("n").asNode()))
//                    .stream()
                        .map(f -> {
                            f.matchSignal(rawFeatures, timestamp);
                            return f.transferFaultToMap();
                        })
                        .filter(map -> (boolean) map.get("isMatched") || (boolean) map.get("isUnknown"))
                        .collect(Collectors.toList());
//        }
    }


    @GET
    @Path("/last/signal/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<? extends Map> lastSignals(@PathParam("timestamp") long timestamp) {
        List<? extends Map> result = Utils.getLastRecords("features",
                Filters.and(
                        Filters.exists("signalType", false), Filters.gt("timestamp", timestamp - 86400000l))
                        .toBsonDocument(null, MongoClient.getDefaultCodecRegistry()), timestamp);
        return isNull(result, Collections.emptyList());
    }

    @GET
    @Path("/last/kg/signal/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Signal> lastKnowledgeSignals(@PathParam("timestamp") long timestamp) {
        List<Document> feature = (List<Document>) lastFeatures(timestamp);
        List<Document> signals = (List<Document>) lastSignals(timestamp);
        List<Signal> rawFeatures = feature.stream()
                .map(SignalUtil::transferFeatureToKG).collect(Collectors.toList());
        List<Signal> rawSignals = signals.stream()
                .map(SignalUtil::transferSignalToKG).collect(Collectors.toList());

        return ListUtils.union(rawFeatures, rawSignals);
    }

    private static <T> T isNull(T val, T orElse) {
        if (val == null) {
            return orElse;
        }
        return val;
    }

    //http://localhost:8082/rest/work/signal/{{machine_id}}/1535972904975
    @GET
    @Path("/last/feature/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<? extends Map> lastFeatures(@PathParam("timestamp") long timestamp) {
        return isNull(Utils.getLastRecords("features",
                Filters.and(
                        Filters.exists("signalType", true), Filters.gt("timestamp", timestamp - 86400000l))
                        .toBsonDocument(null, MongoClient.getDefaultCodecRegistry()), timestamp),
                Collections.emptyList());
    }

    @GET
    @Path("/last/kg/match/{timestamp}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<? extends Map> lastMatch(@PathParam("timestamp") long timestamp) {
        return isNull(Utils.getLastRecords("conclusion", null, timestamp), Collections.emptyList());
    }

//    private Map transferFaultToMap(List<Signal> signalList, Fault fault, long timestamp) {
//        Map resultMap = new HashMap();
//        fault.matchSignal(signalList, timestamp);
//        boolean isTrue = fault.isMatched();
//        resultMap.put("isMatched",isTrue);
//        resultMap.put("isUnknown",fault.isUnKnown());
//        resultMap.put("id",fault.getId());
//        resultMap.put("matched",fault.getSubMatched());
//        resultMap.put("unknown",fault.getSubUnknown());
//        if(isTrue){
//            resultMap.put("outNumber",fault.generateNumber());
//            resultMap.put("faultName",fault.getString("name"));
//        }
//        return resultMap;
//    }

    /**
     * 调用函数
     *
     * @param funcName 函数名
     * @param argsJson 参数数组 [...]
     * @return {result:resultValue}
     */
    @POST
    @Path("/call/{funcName}")
    public Map<String, Object> callFunction(@PathParam("funcName") String funcName, @FormParam("content") String argsJson) {

        return Collections.singletonMap("result", null);
    }

    @POST
    @Path("/query/{collections}")
    @Produces(MediaType.APPLICATION_JSON)
    @Consumes({MediaType.APPLICATION_FORM_URLENCODED})
    /**
     * 根据模版查询
     */
    public List<Map<String, Object>> loadByTemplate(@PathParam("collections") String collections, @FormParam("content") String content) throws IOException {
        MongoDatabase database = Utils.getDatabase("fault");
        MongoCollection<Document> configures = database.getCollection(collections, Document.class);
        /**
         * content 为查询模版，json格式
         */
//        String content = IOUtils.toString(request.getInputStream(),"utf8");
        Document document = Utils.parse(content);
        if (document.containsKey("_id")) {
            String _id = document.getString("_id");
            document.put("_id", new ObjectId(_id));
        }
        return StreamSupport.stream(
                configures.find(Utils.toBson(document))
                        .map(d -> {
                            d.put("_id", d.getObjectId("_id"));
//                            return d.toJson(JSON_WRITER_SETTINGS);
                            return d;
                        })
                        .spliterator(), true).collect(Collectors.toList());
    }


    @GET
    @Path("/start/{unit}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map startWork(@PathParam("unit") String unit) {
//        workTimer.cancel();
        synchronized (workUnits) {
            workUnits.add(unit);
        }
//        workTimer.schedule(workTask,0,250);
        return Collections.singletonMap("state", true);
    }

    @GET
    @Path("/stop/{unit}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map stopWork(@PathParam("unit") String unit) {
//        workTimer.cancel();
        synchronized (workUnits) {
            workUnits.remove(unit);
        }
//        workTimer.schedule(workTask,0,250);
        return Collections.singletonMap("state", true);
    }

    @GET
    @Path("/state/{unit}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map workStatus(@PathParam("unit") String unit) {
        boolean contains = workUnits.stream().filter(x -> StringUtils.equals(x, unit)).findAny().isPresent();
        return Collections.singletonMap("isRun", contains);
    }


    private static Timer workTimer = new Timer();
    private static Set<String> workUnits = new HashSet<>();
    private static TimerTask workTask = new TimerTask() {
        MongoDatabase database = Utils.getDatabase("fault");
        MongoCollection<Document> configures = database.getCollection("configure", Document.class);
        MongoCollection<Document> conclusions = database.getCollection("conclusion", Document.class);
        WorkService workService = new WorkService();

        @Override
        public void run() {
            long timestamp = new Date().getTime();

            synchronized (workUnits) {
                for (String workUnit : workUnits) {
                    workService.generateSignals(workUnit, timestamp);
                    StreamSupport.stream(
                            configures.find(Filters.and(Filters.eq("name", workUnit), Filters.eq("type", "package")))
                                    .sort(Sorts.ascending("processStep")).spliterator(), true)
                            .map(doc -> doc.getObjectId("_id"))
                            .forEach(chain -> workService.doChain(chain, timestamp));
                }
            }
            List<Signal> rawFeatures = workService.getKnowledgeSignals(timestamp);
//            try (NeoDriver driver = new NeoDriver()){
            GraphData graphData = GraphData.getInstance();
            List<Map> results =
                    graphData.getIdSetByLabel("Fault").stream().map(graphData::getById)
                            .map(node1 -> Fault.loadById(node1.id()))
//                        driver.query("match (n:Fault) return n")
//                        .list(record -> new Fault(record.get("n").asNode()))
//                        .stream()
                            .filter(fault -> fault.getOutId() != 0)
                            .filter(fault -> fault.matchSignal(rawFeatures, timestamp))
                            .map(Fault::transferFaultToMap)
                            .peek(map -> map.put("timestamp", timestamp))
                            .collect(Collectors.toList());
            Utils.persistMany(conclusions, results);
//            }
        }
    };

    static {
        String configStr = Neo4jConfigurationUtil.loadConfigure("configure.json");
        JSONObject json = JSONObject.fromObject(configStr);
        long work_span = 10000;
        String WORK_SPAN = json.optString("WORK_SPAN", "10000");
        try {
            work_span = Integer.parseInt(WORK_SPAN);
            if (work_span <= 500) {
                work_span = 500;
            }
        } catch (NumberFormatException ex) {
            System.err.println("配置文件不正确，" + WORK_SPAN + "应当为大于500的整数。 ");
        }
        workTimer.schedule(workTask, 0, work_span);
    }

    /**
     * 用于信号展示
     */
    private static Timer portTimer = new Timer();
    private static String portId = null;
    private static long offsetMs = 0;

    private static TimerTask portTask = new TimerTask() {
        private static final String EXCHANGE_NAME = "TMM_AD_MessagingExchanger";
        ConnectionFactory factory = null;
        Connection connection = null;
        Channel channel = null;
        ObjectMapper mapper = new ObjectMapper();

        {
            factory = new ConnectionFactory();
            factory.setHost(mqHost);
            factory.setPort(mqPort);
            factory.setUsername(mqUser);
            factory.setPassword(mqPassword);

            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            long timestamp = new Date().getTime() - offsetMs;
//            System.out.println(new Date(timestamp)+","+portId);
            if (StringUtils.isNotBlank(portId)) {
                List<Document> signals = Utils.getLastRecords("features",
                        Filters.and(
                                Filters.exists("signalType", false),
                                Filters.eq("portId", portId)
                        ).toBsonDocument(null, MongoClient.getDefaultCodecRegistry()), timestamp);
                if (signals != null && signals.size() > 0) {
                    String message = null;
                    Document document = signals.get(0);
                    List<Double> values = (List<Double>) document.get("value", new ArrayList<>());
                    if (values != null && values.size() > 0)
                        try {
                            message = mapper.writeValueAsString(
                                    new SignalGroup("Monitor")
                                            .addSignal(0, values.get(0)));
                            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                        } catch (JsonProcessingException e) {
                            e.printStackTrace();
                            this.cancel();
                        } catch (IOException e) {
                            e.printStackTrace();
                            this.cancel();
                        }
                }
            }
        }
    };

    static {
        portTimer.scheduleAtFixedRate(portTask, 0, 500);
    }

    @GET
    @Path("/changePort/{id}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map changePort(@PathParam("id") String signalId,
                          @QueryParam("offset") @DefaultValue("0") long offset/*时间偏移 单位ms*/) {
//        portTimer.cancel();
        portId = signalId;
        offsetMs = offset;
        return Collections.singletonMap("_id", signalId);
    }

    private static TendencySourceQueue<Double> tendencyqueue = new TendencySourceQueue<>(20);


    @GET
    @Path("/tendency/{id}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map tendency(@PathParam("id") String signalId,
                        @QueryParam("offset") @DefaultValue("0") long offset/*时间偏移 单位ms*/) {
        //TODO 根据信号id生成趋势
        long timestamp = new Date().getTime() - offsetMs;
        RegressionLine line = new RegressionLine();
        if (StringUtils.isNotBlank(portId)) {
            List<Document> signals = Utils.getLastRecords("features",
                    Filters.and(
                            Filters.exists("signalType", false),
                            Filters.eq("portId", portId)
                    ).toBsonDocument(null, MongoClient.getDefaultCodecRegistry()), timestamp);
            if (signals != null && signals.size() > 0) {
                Document document = signals.get(0);
                List<Double> values = (List<Double>) document.get("value", new ArrayList<>());
                tendencyqueue.add(values.get(0));

                //System.out.println(tendencyqueue);
                if (tendencyqueue.size() == tendencyqueue.getLimit()) {
                    Double[] tendencyValues = tendencyqueue.toArray(new Double[0]);
                    //System.out.println(Arrays.toString(tendencyValues));
                    //向直线中加点
                    for (int i = 1; i <= tendencyValues.length; i++) {
                        line.addDataPoint(new DataPoint(i, tendencyValues[i - 1]));
                    }

                }
            }
        }
        return Collections.singletonMap("result", line.getA1()/**/);
    }


    @GET
    @Path("/orbitList/{id}")
    @Produces(MediaType.APPLICATION_JSON)
    public List<Map> orbitTrackList(@PathParam("id") String portId,
                                    @DefaultValue("0") @QueryParam("begin") long beginTimeStamp,
                                    @DefaultValue("0") @QueryParam("end") long endTimeStamp) {
        List<List<Double>> data = OrbitTrack.getOrbitHistory(portId, beginTimeStamp, endTimeStamp);
        return data.stream().map(lst -> {
            Map map = new HashMap();
            map.put("x", lst.get(0));
            map.put("y", lst.get(1));
            return map;
        }).collect(Collectors.toList());
    }

    @GET
    @Path("/orbit/{id}")
    @Produces(MediaType.APPLICATION_JSON)
    public Map orbitTrack(@PathParam("id") String portId,
                          @DefaultValue("0") @QueryParam("begin") long beginTimeStamp,
                          @DefaultValue("0") @QueryParam("end") long endTimeStamp) {

        List<List<Double>> data = OrbitTrack.getOrbitHistory(portId, beginTimeStamp, endTimeStamp);

        List<Double> tranX = data.stream().mapToDouble(lst -> lst.get(0)).boxed().collect(Collectors.toList());
        List<Double> tranY = data.stream().mapToDouble(lst -> lst.get(1)).boxed().collect(Collectors.toList());
        List<List<Double>> transferedData = Arrays.asList(tranX, tranY);

        try {
            String result = setdata.ZXGJ.ZhouXinGuiJi(transferedData);
            return Collections.singletonMap("result", result);
        } catch (ClassNotFoundException | IOException e) {
            e.printStackTrace();
            return Collections.singletonMap("result", "ERROR");
        }
    }

//    public static void main(String[] args) throws IOException {
//        portId="5acca9729f72cc116cec46a7";
//        Date date = new Date();
//        date.setHours(16);
//        date.setMinutes(11);
//        offsetMs = new Date().getTime() - date.getTime();
//        System.in.read();
//    }

    static {
        try {
            Class.forName("cn.edu.hit.core.transfer.TransferLoader");
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}
